package com.j.app.function;/**
 * @ClassName DimSinkFunction
 * @Description TODO
 * @date 2024/3/5 19:43
 * @Author JG
 */

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.druid.pool.DruidPooledConnection;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.j.utils.DruidDSUtil;
import com.j.utils.PhoenixUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

/**
 * @ClassName DimSinkFunction
 * @Description TODO
 * @date 2024/3/5 19:43
 * @Author JG
 */
public class DimSinkFunction extends RichSinkFunction<JSONObject> {
    private static DruidDataSource druidDataSource=null;

    @Override
    public void open(Configuration parameters) throws Exception {
        druidDataSource= DruidDSUtil.createDataSource();
    }

    @Override
    public void invoke(JSONObject value, Context context) throws Exception {
        //获取连接
        DruidPooledConnection connection = druidDataSource.getConnection();
        //写出数据
        String sinktable = value.getString("sinkTable");
        JSONObject data = value.getJSONObject("data");
        try {
            PhoenixUtil.insertValues(connection,sinktable,data);
        } catch (Exception e) {
            e.printStackTrace();
        }
        //归还连接
        connection.close();

    }
}
